package com.cqw.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cqw.constant.MqConstant;
import com.cqw.domain.Area;
import com.cqw.domain.HttpClientResult;
import com.cqw.domain.SysUser;
import com.cqw.mapper.one.AreaMapper;
import com.cqw.utils.FirstLetterUtil;
import com.cqw.utils.HttpClientUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Author: caoqingwen
 * @CreateTime: 2021-04-15 14:12
 * @Description: 消息接受者
 */
@Component
@Slf4j
public class MqReceive {


    @Value("${gaode.area-url}")
    private String url;

    @Value("${gaode.key}")
    private String key;

    @Resource
    private AreaMapper areaMapper;


    @RabbitListener(queues = MqConstant.AREA_QUEUE)
    public void receiveArea(Message message, Channel channel) {
        log.info("初始化区域数据");
        try {
            Map<String, String> map = new HashMap<>();
            map.put("key", key);
            map.put("subdistrict", "3");
            HttpClientResult httpClientResult = HttpClientUtils.doGet(url, map);
            //区域JSON串
            String content = httpClientResult.getContent();
            if (StringUtils.isNotBlank(content)) {
                JSONObject jsonObject = JSON.parseObject(content);
                String countryListStr = jsonObject.getString("districts");
                List<Area> countryList = JSON.parseArray(countryListStr, Area.class);
                Area country = countryList.get(0);
                //获取所有的省份
                List<Area> provinceList = country.getDistricts();
                //排序
                provinceList = provinceList.stream()
                        .sorted(Comparator.comparing(Area::getAdCode))
                        .collect(Collectors.toList());
                //处理数据关系
                recursion(provinceList, null, "");
                log.info("初始化区域数据成功");
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.info("初始化区域数据失败");
        }
    }


    /**
     * @param parentList   父级行政单位数据
     * @param parentAdCode 父级的adCode
     */
    private void recursion(List<Area> parentList, String parentAdCode, String parentName) {
        if (CollectionUtils.isEmpty(parentList)) {
            return;
        }

        String fullName = "";

        //处理cityCode为【】，街道adCode于上一级相同的情况
        for (int i = 0; i < parentList.size(); i++) {
            Area parent = parentList.get(i);

            //处理首字母
            String firstLetter = FirstLetterUtil.getFirstLetter(parent.getName().substring(0, 1));
            parent.setInitial(firstLetter.toUpperCase());

            if ("[]".equals(parent.getCityCode())) {
                parent.setCityCode("");
            }
            if ("street".equals(parent.getLevel())) {
                String adCode = parent.getAdCode().concat(String.format("%02d", i + 1));
                parent.setAdCode(adCode);
            }

            if ("province".equals(parent.getLevel())) {
                parent.setParentCode("0");
                //省级的全名就是他的名字
                parent.setFullName(parent.getName());
                fullName = parent.getName();
            } else {
                parent.setParentCode(parentAdCode);
                fullName = parentName + parent.getName();
                parent.setFullName(fullName);
            }

            int insert = areaMapper.insert(parent);

            if (insert > 0 && !CollectionUtils.isEmpty(parent.getDistricts())) {
                recursion(parent.getDistricts(), parent.getAdCode(), fullName);
            }
        }
    }


    //测试mq消费者
    @RabbitListener(queues = MqConstant.CE_SHI_QUEUE)
    public void ceShi(String message) {
        SysUser user = JSON.parseObject(message, SysUser.class);
        System.out.println(user);
    }

    //测试mq消费者
    @RabbitHandler
    @RabbitListener(queues = MqConstant.LOG_QUEUE1)
    public void ceShi1(String message) {
        SysUser user = JSON.parseObject(message, SysUser.class);
        System.out.println(user);
    }


    //===============下面是对几种模型的测试start


    /**
     * 直连模式
     *
     * @param message
     */
    @RabbitListener(queues = "hello")
    public void hello1(String message) {
        System.out.println(message);
        System.out.println("消费者1消费成功");
    }

    @RabbitListener(queues = "hello")
    public void hello2(String message) {
        System.out.println(message);
        System.out.println("消费者2消费成功");
    }


    /**
     * 广播模式：两个消费者都能收到消息
     */
    @RabbitListener(queues = "fanoutQueue1")
    public void fanout1(String message) {
        System.out.println("消费者1消费：" + message);
    }

    @RabbitListener(queues = "fanoutQueue2")
    public void fanout2(String message) {
        System.out.println("消费者2消费：" + message);
    }

    /**
     * direct模式：根据路由key，发送到不同的队列中
     */
    @RabbitListener(queues = "directQueue1")
    public void direct1Info1(Message message, Channel channel) throws IOException {
        byte[] body = message.getBody();
        String s = new String(body);
        //消息投递id，是MQ服务器产生的
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        //multiple: 是否批量签收
        channel.basicAck(deliveryTag, false);
        System.out.println("info消费者1：" + s);
    }

    @RabbitListener(queues = "directQueue1")
    public void direct1Info11(Message message, Channel channel) throws IOException {
        byte[] body = message.getBody();
        String s = new String(body);
        //消息投递id，是MQ服务器产生的
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        //multiple: 是否批量签收
        channel.basicAck(deliveryTag, false);
        System.out.println("info消费者2：" + s);
    }

    @RabbitListener(queues = "directQueue2")
    public void direct1Info2(String str, Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("error消费者1：" + str);
    }

    @RabbitListener(queues = "directQueue2")
    public void direct1Info21(String str, Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("error消费者2：" + str);
    }

    /**
     * topic
     */
    @RabbitListener(queues = "topicQueue1")
    public void topic11(String str, Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("user.*消费者1：" + str);
    }

    @RabbitListener(queues = "topicQueue1")
    public void topic12(String str, Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("user.*消费者2：" + message);
    }

    @RabbitListener(queues = "topicQueue2")
    public void topic21(String str, Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("user.#消费者1：" + str);
    }

    @RabbitListener(queues = "topicQueue2")
    public void topic22(String str, Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("user.#消费者2：" + str);
    }

    @RabbitListener(queues = "deadly.new.queue")
    public void deadly(String str, Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        System.out.println("deadly消费者：" + str);
    }

}
